

#ifndef PARALLELMEMORYMANAGER_H_
#define PARALLELMEMORYMANAGER_H_


#include <boost/unordered_map.hpp>
#include <algorithm>
#include <vector>
#include <queue>
#include "../Common/Consts.h"
#include "Components/Encoder.h"
#include "Components/Collector.h"
#include "Memory Tables/MemoryTable.h"
#include "../Util/Util.h"

using namespace std;


class ParallelMemoryManager : public MemoryManager {

public:
	/**
	 * Creates the Memory Manager
	 */
	ParallelMemoryManager(MemoryTable *parMemoryTable, int subscriptionThreshold, int parNumClusters, float parLastThreshold);

	virtual ~ParallelMemoryManager();

	/**
	 * Starts the Memory Manager
	 */

	void addEvent(PubPkt *pkg);

	void addSubscription(BackSubPkt *pkg);

	void queryEventSet(BackSubPkt *pkg, Outbox &outbox);

	/**
	 * the thread to read from the queue of input subscriptions and process them to store informations.
	 */
	void manageSubscription();

	/**
	 * Start the thread to read from the queue of input subscriptions and process them to store informations and to store events themselves
	 */
	void manageEvents();

	/**
	 * Start the thread to update the table of informations for the storage
	 */
	void updateTable();


private:


	queue<PubPkt *> inputEventQueue;					// Input queue for received events
	queue<BackSubPkt *> inputSubscriptionQueue;				// Input queue for received subscriptions

	pthread_cond_t *inputEventCond;						// Condition variable for the input queue
	pthread_mutex_t *inputEventMutex;					// Mutex for the input queue

	pthread_cond_t *inputSubscriptionCond;				// Condition variable for the input queue
	pthread_mutex_t *inputSubscriptionMutex;			// Mutex for the input queue

	pthread_t *eventThread;								// Thread managing events
	pthread_t *subscriptionThread;						// Thread managing subscriptions
	pthread_t *updatingThread;							// Thread managing the memory table

	Collector *collector;

	Encoder *encoder;
	pthread_mutex_t *encoderMutex;

	typedef struct SubscribtionInfoStruct {
		vector<float> windowsList;
		int countLast;
	} SubscribtionInfo;

	boost::unordered_map<string, SubscribtionInfo> subscriptionMap;
	boost::unordered_map<string, int> eventMap;

	pthread_mutex_t *subscriptionMapMutex;
	pthread_mutex_t *eventMapMutex;

	MemoryTable *memoryTable;
	pthread_mutex_t *memoryTableMutex;

	int collectorThreshold;
	float lastThreshold;
	bool memoryTableToReset;
	pthread_mutex_t *memoryTableToResetMutex;
};

namespace parallel_manager {


/**
 * Start the thread to read from the queue of input events and sends them to the T-Rex engine.
 */
void * startEventThread(void *input);

/**
 * Start a new thread to handle messages coming from a connected client.
 * Each message is appended to the input queue, or dropped if the queue is full.
 */
void * startSubscriptionThread(void *input);

/**
 * Start a new thread to send messages to connected clients.
 */
void * startUpdatingThread(void *input);

}


#endif /* PARALLELMEMORYMANAGER_H_ */
